#include <threadpool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include <http_serv_common.h>
#include <unistd.h>
#include <strings.h>

void *
thread_routine(void *arg)
{
	DEBUG_PRINT("thread 0x%08x is starting\n",(int)pthread_self());
	struct timespec abstime;
	int timeout;
	//等待任务，等待任务后就执行
	//消费线程，判断链表是否空的
	//条件来等待
	threadpool_t *pool = (threadpool_t *)arg;
	while(1) {
		timeout = 0;
		//规范：1.加锁 2.解锁
		condition_lock(&pool->ready);
		//等待队列不空或者线程池销毁通知
		pool->idle++;//表明有个线程要等待
		while(NULL == pool->first && !pool->quit) {
			clock_gettime(CLOCK_REALTIME,&abstime);
			abstime.tv_sec += THREAD_POOL_TASK_TIMEOUT;
			int status = condition_timedwait(&pool->ready,
							&abstime);
			if(ETIMEDOUT == status) {
				DEBUG_PRINT("thread 0x%08x is wait timed out\n",
							(int)pthread_self());
				timeout = 1;
				break;
			}
		}
		pool->idle--;//等待到条件

		if(NULL != pool->first) {
			//从队头取出任务处理
			task_t *t = pool->first;
			pool->first = t->next;
			//执行任务前需要解锁以便生产者进程能够
			//往队列中添加任务，其它消费者线程等待
			//任务。
			condition_unlock(&pool->ready);
			t->run(t->arg);//执行任务要时间
			free(t);
			condition_lock(&pool->ready);
		}
		//如果等待到线程池销毁通知
		if(pool->quit && NULL == pool->first) {
			pool->counter--;
			if(0 == pool->counter) {
				condition_signal(&pool->ready);
			}
			condition_unlock(&pool->ready);
			break;
		}
		//如果超时
		if(timeout && NULL == pool->first) {
			pool->counter--;
			condition_unlock(&pool->ready);
			break;
		}
		condition_unlock(&pool->ready);
	}
	DEBUG_PRINT("thread 0x%08x is exiting\n",(int)pthread_self());
	return NULL;
}

//初始化线程池
void 
threadpool_init(threadpool_t *pool,int threads)
{
	condition_init(&pool->ready);
	//队列的初始化
	pool->first = NULL;
	pool->last = NULL;
	pool->counter = 0;
	pool->idle = 0;
	pool->max_threads = threads;
	pool->quit = 0;
}
//往线程池中添加任务
void 
threadpool_add_task(threadpool_t *pool,void *(*run)(void *arg),void *arg)
{
	//首先是生成一个任务
	task_t *newtask = (task_t *)malloc(sizeof(task_t));
	newtask->run = run;
	newtask->arg = arg;
	newtask->next = NULL;
	//链表插入尾部
	//对互斥锁进行加锁
	condition_lock(&pool->ready);
	if(NULL == pool->first) {
		pool->first = newtask;
	} else {
		pool->last->next = newtask;
	}
	pool->last = newtask;
	//如果有等待线程就唤醒其中一个
	if(pool->idle > 0) {
		condition_signal(&pool->ready);
	} else if(pool->counter < pool->max_threads) {
		//创建新的线程
		pthread_t tid;
		pthread_create(&tid,NULL,thread_routine,pool);
		pool->counter++;
	}
	condition_unlock(&pool->ready);
}
//销毁线程池
void threadpool_destroy(threadpool_t *pool)
{
	//等待所有线程将任务执行完毕后再销毁
	if(pool->quit) {
		return;
	}
	condition_lock(&pool->ready);
	pool->quit = 1;
	if(pool->counter > 0) {
		//这时才有必要销毁
		//要判断线程是否都在等待状态，如果不在
		//等待状态就无法接收广播
		if(pool->idle > 0) {
			condition_broadcast(&pool->ready);
		}
		//处于执行任务状态中的线程，不会收到广播
		//线程池需要等待执行任务状态中的线程全部
		//退出
		while(pool->counter > 0) {
			condition_wait(&pool->ready);
		}
	}
	condition_unlock(&pool->ready);
	condition_destroy(&pool->ready);
}

typedef	struct
{
	//单个任务所需要的内存长度
	uint32_t s_task_l;
	//线程数
	int32_t threads;
	//空间使用状态数组
	bool *th_use_stat;
	//单个任务内存各自的起
	//始地址，和线程数一样
	uint8_t **s_task_start_ad;
	//总的地址
	uint8_t *task_mem;
} mem_poll_t;

mem_poll_t mpt;
//建立一个仓库机制和线程数匹配，主要用途是为每个线程
//传入的任务分配一个唯一的内存空间，减少原来为了防止
//临界资源问题而每次分配内存复制数据的情况。
void threadpool_task_mem_pool_init(int32_t threads,uint32_t single_task_len)
{
	mpt.s_task_l = single_task_len;
	mpt.threads = threads;
	#ifdef _USE_MMAP
	mpt.task_mem = (uint8_t *)mmap(NULL,threads * single_task_len,
			PROT_READ | PROT_WRITE,MAP_SHARED | MAP_ANON,-1,0);
	if(MAP_FAILED == mpt.task_mem) {
		ERR_EXIT("mmap");
	}
	#else
	mpt.task_mem = (uint8_t *)malloc(threads * single_task_len);
	#endif
	bzero(mpt.task_mem,threads * single_task_len);
	mpt.th_use_stat = (bool *)malloc(sizeof(bool) * threads);
	memset(mpt.th_use_stat,false,sizeof(bool) * threads);
	mpt.s_task_start_ad = (uint8_t **)malloc(sizeof(uint8_t *) * threads);
	bzero(mpt.s_task_start_ad,sizeof(uint8_t *) * threads);
	int32_t i;
	for(i = 0;i < threads;++i) {
		mpt.s_task_start_ad[i] = mpt.task_mem \
						+ single_task_len * i;
	}
}
//获取一块内存用来存放任务数据
void *threadpool_malloc_task_mem()
{
	//首先要找到一块没有使用的内存
	int i;
	for(i = 0;i < mpt.threads;++i) {
		if(!mpt.th_use_stat[i]) {
			break;
		}
	}
	mpt.th_use_stat[i] = true;
	return (void *)mpt.s_task_start_ad[i];
}
bool threadpool_free_task_mem(void *task_mem)
{
	//找到对应的状态，将其置为未使用状态
	int i;
	for(i = 0;i < mpt.threads;++i) {
		if(task_mem == mpt.s_task_start_ad[i]) {
			mpt.th_use_stat[i] = false;
			return true;
		}
	}
	return false;
}
//释放这个内存仓库
void threadpool_task_mem_pool_destroy(void)
{
	if(NULL != mpt.task_mem) {
	#ifdef _USE_MMAP
		munmap(mpt.task_mem,mpt.s_task_l * mpt.threads);
	#else
		free(mpt.task_mem);
	#endif
	}
	if(NULL != mpt.th_use_stat) {
		free(mpt.th_use_stat);
	}
	if(NULL != mpt.s_task_start_ad) {
		free(mpt.s_task_start_ad);
	}
}
